package com.xbongbong.pro.rabbitmq.consumer;

import com.alibaba.fastjson.JSON;
import com.xbongbong.paas.redis.PaasRedisHelper;
import com.xbongbong.paas.redis.RedisPrefixConstant;
import com.xbongbong.paas.redis.RedisTimeConstant;
import com.xbongbong.paas.toolbox.wrap.XbbResponse;
import com.xbongbong.pro.config.ProBaseConfig;
import com.xbongbong.pro.connector.pojo.dto.ConnectorHistoryDataInitDTO;
import com.xbongbong.pro.connector.pojo.dto.ConnectorRabbitMqDTO;
import com.xbongbong.pro.connector.pojo.vo.ConnectorRabbitMqVO;
import com.xbongbong.pro.rabbitmq.binders.ConnectorConsumerBinder;
import com.xbongbong.pro.rabbitmq.constant.ConnectorBinderConstant;
import com.xbongbong.pro.rabbitmq.feign.ConnectorFeignClient;
import com.xbongbong.saas.enums.OperateTypeEnum;
import com.xbongbong.saas.enums.XbbRefTypeEnum;
import feign.RetryableException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.cloud.stream.annotation.EnableBinding;
import org.springframework.cloud.stream.annotation.StreamListener;
import org.springframework.context.i18n.LocaleContextHolder;

import javax.annotation.Resource;
import java.util.Iterator;
import java.util.List;
import java.util.Locale;
import java.util.Objects;

/**
 * 连接器消费者
 * 创建时间 2020/12/25 4:59 PM
 * 修改时间 2020/12/25 4:59 PM
 * @author chy
 */
@EnableBinding(ConnectorConsumerBinder.class)
public class ConnectorListener {
    private static Logger LOG = LoggerFactory.getLogger(ConnectorListener.class);
    @Resource
    private ProBaseConfig proBaseConfig;
    @Resource
    private ConnectorFeignClient connectorFeignClient;
    @Resource
    private PaasRedisHelper paasRedisHelper;
    public void triggerSend(ConnectorRabbitMqDTO connectorRabbitMqDTO) {
        XbbResponse<ConnectorRabbitMqVO> response;
        try {
            String corpid = connectorRabbitMqDTO.getCorpid();
            XbbRefTypeEnum xbbRefTypeEnum = connectorRabbitMqDTO.getXbbRefTypeEnum();
            List<Long> dataIdList = connectorRabbitMqDTO.getDataIdList();
            String operate = connectorRabbitMqDTO.getOperate();
            Iterator<Long> it = dataIdList.iterator();
            while (it.hasNext()) {
                Long dataId = it.next();
                if (!checkConnectionConcurrentLock(corpid, xbbRefTypeEnum, dataId, operate)) {
                    it.remove();
                    LOG.info("被锁了---dataId:" + dataId);
                }
            }
            if (dataIdList.isEmpty()) {
                return;
            }
            connectorRabbitMqDTO.setDataIdList(dataIdList);
            Locale locale = LocaleContextHolder.getLocale();
            String lang = locale.toString();
            connectorRabbitMqDTO.setXbbCode(proBaseConfig.getXbbCode());
            response = connectorFeignClient.triggerSend(connectorRabbitMqDTO, lang);
            LOG.info("triggerSend message=" + JSON.toJSONString(connectorRabbitMqDTO) + "\n response=" + JSON.toJSONString(response));
        } catch (RetryableException e) {
            LOG.error("ConnectorListener.triggerSend RetryableException case by Read timed out");
        } catch (Exception e) {
            LOG.error("ConnectorListener.triggerSend Exception", e);
        }
    }

    public void historyDataInit(ConnectorHistoryDataInitDTO connectorHistoryDataInitDTO) {
        XbbResponse<ConnectorRabbitMqVO> response;
        try {
            Locale locale = LocaleContextHolder.getLocale();
            String lang = locale.toString();
            connectorHistoryDataInitDTO.setXbbCode(proBaseConfig.getXbbCode());
            response = connectorFeignClient.historyDataInit(connectorHistoryDataInitDTO, lang);
            LOG.info("pullData message=" + JSON.toJSONString(connectorHistoryDataInitDTO) + "\n response=" + JSON.toJSONString(response));
        } catch (RetryableException e) {
            LOG.error("ConnectorListener.triggerSend RetryableException case by Read timed out");
        } catch (Exception e) {
            LOG.error("ConnectorListener.triggerSend Exception", e);
        }
    }

    @StreamListener(value = ConnectorBinderConstant.CONNECTOR_INPUT)
    public void connectorInput(ConnectorRabbitMqDTO connectorRabbitMqDTO) {
        LOG.info("connectorInput Received: " + JSON.toJSONString(connectorRabbitMqDTO));
        triggerSend(connectorRabbitMqDTO);
    }

    @StreamListener(value = ConnectorBinderConstant.CONNECTOR_HISTORY_DATA_INIT_INPUT)
    public void connectorHistoryDataInitInput(ConnectorHistoryDataInitDTO connectorHistoryDataInitDTO) {
        LOG.info("connectorPullDataInput Received: " + JSON.toJSONString(connectorHistoryDataInitDTO));
        historyDataInit(connectorHistoryDataInitDTO);
    }

    /**
     * 获取连接器并发锁 若没有被占用则返回true并构建锁，若已被占用则返回false
     * 并发锁时间：默认三秒
     * @param corpid 公司id
     * @param xbbRefTypeEnum 业务类型
     * @param dataId 数据id
     * @param operate 操作类型
     * @return
     */
    public boolean checkConnectionConcurrentLock(String corpid, XbbRefTypeEnum xbbRefTypeEnum, Long dataId, String operate) {
        String lockStr = corpid + "_" + xbbRefTypeEnum.getCode() + "_" + dataId + "_" + operate;
        return paasRedisHelper.checkConcurrentLock(RedisPrefixConstant.CONNECTOR, lockStr, RedisTimeConstant.MINI_DURATION, true);
    }
}
